package com.chukun.multi.producerconsumer.workstealing;

import java.util.Objects;
import java.util.concurrent.BlockingDeque;

/**
 * 基于产品窃取的队列的实现
 * @param <T>
 */
public class WorkStealingChannel<T> implements WorkStealingEnabledChannel<T> {

    // 受管队列
    private final BlockingDeque<T>[] managedQueues;

    public WorkStealingChannel(BlockingDeque<T>[] managedQueues) {
        this.managedQueues = managedQueues;
    }

    /**
     * 实现产品窃取的方法
     * @param preferredQueue
     * @return
     * @throws InterruptedException
     */
    @Override
    public T take(BlockingDeque<T> preferredQueue) throws InterruptedException {

        // 优先从指定的受管队列中取出产品
        BlockingDeque<T> targetQueue = preferredQueue;
        T product = null;

        if (!Objects.isNull(targetQueue)) {
            product = targetQueue.poll();
        }

        int queueIndex = -1;
         // 顺序获取
        while (Objects.isNull(product)) {
            // 顺序 "窃取" 其他受管队列的产品
            queueIndex = (queueIndex + 1) % managedQueues.length;
            targetQueue = managedQueues[queueIndex];
            // 从队尾获取元素
            product = targetQueue.pollLast();
            // 若顺序得到的队列，与优先选择的队列相同，则退出
            if (preferredQueue == targetQueue) {
                break;
            }
        }

        // 随机获取
        if (Objects.isNull(product)) {
            // 随机 "窃取" 其他受管队列的产品
            queueIndex = (int) System.currentTimeMillis() % managedQueues.length;
            targetQueue = managedQueues[queueIndex];
            product = targetQueue.takeLast();
            System.out.println("steal product from " + queueIndex + " ： " + product);
        }
        return product;
    }

    @Override
    public T take() throws InterruptedException {
        return take(null);
    }

    /**
     * 放入产品的方式,利用产品的product的hashCode方法 mod 受管队列的长度
     * 获取要放入的队列的index，放入产品
     * @param product
     * @throws InterruptedException
     */
    @Override
    public void put(T product) throws InterruptedException {
        int targetIndex = product.hashCode() % managedQueues.length;
        managedQueues[targetIndex].put(product);
    }
}
